/*
 * Copyright (C) 2008 by The Regents of the University of California
 * Redistribution of this file is permitted under the terms of the GNU
 * Public License (GPL).
 *
 * @author Junghoo "John" Cho <cho AT cs.ucla.edu>
 * @date 3/24/2008
 */
 
#include <iostream>
#include "BTreeIndex.h"

using namespace std;

#define DOUT if (0) cout

/*
 * BTreeIndex constructor
 */
BTreeIndex::BTreeIndex()
{
    rootPid = -1;
}

/*
 * Open the index file in read or write mode.
 * Under 'w' mode, the index file should be created if it does not exist.
 * @param indexname[IN] the name of the index file
 * @param mode[IN] 'r' for read, 'w' for write
 * @return error code. 0 if no error
 */
RC BTreeIndex::open(const string& indexname, char mode)
{
    int errorCode = pf.open(indexname, mode);
    if (errorCode == 0)
    {
      char buffer[PageFile::PAGE_SIZE];
      errorCode = pf.read(0, buffer);
      if (errorCode == 0)
      {
        DOUT << "Reading meta state from pf at pid 0." << endl;
        memcpy(&rootPid, &buffer[0], sizeof(rootPid));
        memcpy(&treeHeight, &buffer[0 + sizeof(rootPid)], sizeof(treeHeight));
        memcpy(&minKey, &buffer[0 + sizeof(rootPid) + sizeof(treeHeight)], sizeof(minKey));
      }
      else if (errorCode == RC_INVALID_PID)
      {
        rootPid = 1;
        treeHeight = 1;
        minKey = INT_MAX;
        char buffer[PageFile::PAGE_SIZE];
        memcpy(&buffer[0], &rootPid, sizeof(rootPid));
        memcpy(&buffer[0 + sizeof(rootPid)], &treeHeight, sizeof(treeHeight));
        memcpy(&buffer[0 + sizeof(rootPid) + sizeof(treeHeight)], &minKey, sizeof(minKey));
        DOUT << "Writing initial meta state to pf at pid 0." << endl;
        errorCode = pf.write(0, buffer);
      }
    }
    return errorCode;
}

/*
 * Close the index file.
 * @return error code. 0 if no error
 */
RC BTreeIndex::close()
{
    DOUT << "CLosing pf!" << endl;
    char buffer[PageFile::PAGE_SIZE];
    memcpy(&buffer[0], &rootPid, sizeof(rootPid));
    memcpy(&buffer[0 + sizeof(rootPid)], &treeHeight, sizeof(treeHeight));
    memcpy(&buffer[0 + sizeof(rootPid) + sizeof(treeHeight)], &minKey, sizeof(minKey));
    pf.write(0, buffer);
    return pf.close();
}

/*
 * Insert (key, RecordId) pair to the index.
 * @param key[IN] the key for the value inserted into the index
 * @param rid[IN] the RecordId for the record being inserted into the index
 * @return error code. 0 if no error
 */
RC BTreeIndex::insert(int key, const RecordId& rid)
{
    int overflowKey = -1;
    int siblingPid = -1;
    insertNonLeaf(key, rid, rootPid, 1, overflowKey, siblingPid);
    if (overflowKey != -1)
    {
      // Create new root
      DOUT << "Creating new root!" << endl;
      BTNonLeafNode newRoot;
      int newRootPid = pf.endPid();
      newRoot.read(newRootPid, pf);
      newRoot.initializeRoot(rootPid, overflowKey, siblingPid);
      DOUT << "New Root Pid: " << newRootPid << endl;
      DOUT << "Root Pid: " << rootPid << endl;
      DOUT << "overflowKey: " << overflowKey << endl;
      DOUT << "Sibling Pid: " << siblingPid << endl;
      rootPid = newRootPid;
      treeHeight += 1;
      newRoot.write(newRootPid, pf);
    }
    if (key < minKey)
    {
      DOUT << "Setting new min key!" << endl;
      minKey = key;
    }
    DOUT << "Done inserting!" << endl;
    return 0;
}

RC BTreeIndex::insertNonLeaf(int key, const RecordId& rid, PageId pid, int currentHeight, int& overflowKey,
                             int& siblingPid)
{
  if (currentHeight == treeHeight)
  {
    DOUT << "Going to insert into leaf node!" << endl;
    insertLeaf(key, rid, pid, overflowKey, siblingPid);
    DOUT << "Sibling Key: " << overflowKey << endl;
    DOUT << "Sibling Pid: " << siblingPid << endl;
  }
  else
  {
    DOUT << "Recursively following childPtr to locate insertion point!" << endl;
    int childPid;
    BTNonLeafNode currentNode;
    currentNode.read(pid, pf);
    currentNode.locateChildPtr(key, childPid);
    insertNonLeaf(key, rid, childPid, currentHeight + 1, overflowKey, siblingPid);
    handleOverflowIfNeeded(currentNode, pid, siblingPid, overflowKey);
  }
}

RC BTreeIndex::handleOverflowIfNeeded(BTNonLeafNode node, PageId currentNodePid, PageId& siblingPid, int& overflowKey)
{
    if (overflowKey != -1)
    {
      DOUT << "Handling overflow!" << endl;

      int errorCode = node.insert(overflowKey, siblingPid);

      if (errorCode == RC_NODE_FULL)
      {
        DOUT << "Splitting non-leaf node!" << endl;
        BTNonLeafNode newSibling;
        int newOverflowKey;
        PageId newSiblingPid = pf.endPid();
        newSibling.read(newSiblingPid, pf);
        node.print();
        node.insertAndSplit(overflowKey, siblingPid, newSibling, newOverflowKey);
        node.print();
        newSibling.print();
        newSibling.write(newSiblingPid, pf);
        node.write(currentNodePid, pf);
        siblingPid = newSiblingPid;
        overflowKey = newOverflowKey;
        DOUT << "Done splitting non-leaf!" << endl;
      }
      else
      {
        DOUT << "No carryout overflow!" << endl;
        node.write(currentNodePid, pf);
        overflowKey = -1;
        siblingPid = -1;
      }
    }
    else
    {
      DOUT << "No overflow!" << endl;
      overflowKey = -1;
      siblingPid = -1;
    }
}

RC BTreeIndex::insertLeaf(int key, const RecordId& rid, PageId pid, int& overflowKey, int& siblingPid)
{
  BTLeafNode leaf;
  leaf.read(pid, pf);
  leaf.print();
  overflowKey = -1;
  DOUT << "Inserting key " << key << "into leaf at pid " << pid << endl;
  int errorCode = leaf.insert(key, rid);
  leaf.print();
  if (errorCode == RC_NODE_FULL)
  {
    DOUT << "Leaf node at pid " << pid << " full, splitting!" << endl;
    BTLeafNode sibling;
    siblingPid = pf.endPid();
    sibling.read(siblingPid, pf);
    errorCode = leaf.insertAndSplit(key, rid, sibling, overflowKey);
    DOUT << "Sibling Key: " << overflowKey << endl;
    DOUT << "Sibling Pid: " << siblingPid << endl;
    leaf.write(pid, pf);
    sibling.write(siblingPid, pf);
    leaf.print();
    sibling.print();
  }
  else
  {
    DOUT << "No leaf overflow!" << endl;
    int writeStatus = leaf.write(pid, pf);
    if (writeStatus != 0)
    {
      DOUT << "Error writing!" << endl;
    }
    overflowKey = -1;
    siblingPid = -1;
  }

  return errorCode;
}

/*
 * Find the leaf-node index entry whose key value is larger than or 
 * equal to searchKey, and output the location of the entry in IndexCursor.
 * IndexCursor is a "pointer" to a B+tree leaf-node entry consisting of
 * the PageId of the node and the SlotID of the index entry.
 * Note that, for range queries, we need to scan the B+tree leaf nodes.
 * For example, if the query is "key > 1000", we should scan the leaf
 * nodes starting with the key value 1000. For this reason,
 * it is better to return the location of the leaf node entry 
 * for a given searchKey, instead of returning the RecordId
 * associated with the searchKey directly.
 * Once the location of the index entry is identified and returned 
 * from this function, you should call readForward() to retrieve the
 * actual (key, rid) pair from the index.
 * @param key[IN] the key to find.
 * @param cursor[OUT] the cursor pointing to the first index entry
 *                    with the key value.
 * @return error code. 0 if no error.
 */
RC BTreeIndex::locate(int searchKey, IndexCursor& cursor)
{
    return locateRecursive(searchKey, rootPid, 1, cursor);
}

RC BTreeIndex::locateRecursive(int searchKey, PageId pid, int height, IndexCursor& cursor)
{
    if (height == treeHeight)
    {
      BTLeafNode leaf;
      DOUT << "Looking in leaf node with pid: " << pid << endl;
      int errorCode = leaf.read(pid, pf);
      if (errorCode != 0)
      {
        DOUT << "ERROR: " << errorCode << endl;
        return errorCode;
      }
      cursor.pid = pid;
      leaf.print();
      return leaf.locate(searchKey, cursor.eid);
    }
    else
    {
      DOUT << "Looking in non-leaf now w/ pid: " << pid << endl;
      BTNonLeafNode currentNode;
      currentNode.read(pid, pf);
      int childPid;
      currentNode.locateChildPtr(searchKey, childPid);
      locateRecursive(searchKey, childPid, height + 1, cursor);
    }
}

/*
 * Read the (key, rid) pair at the location specified by the index cursor,
 * and move foward the cursor to the next entry.
 * @param cursor[IN/OUT] the cursor pointing to an leaf-node index entry in the b+tree
 * @param key[OUT] the key stored at the index cursor location.
 * @param rid[OUT] the RecordId stored at the index cursor location.
 * @return error code. 0 if no error
 */
RC BTreeIndex::readForward(IndexCursor& cursor, int& key, RecordId& rid)
{
    int errorCode = 0;
    BTLeafNode leaf;

    errorCode = leaf.read(cursor.pid, pf);
    if (errorCode != 0)
    {
      return errorCode;
    }

    errorCode = leaf.readEntry(cursor.eid, key, rid);
    if (errorCode != 0)
    {
      return errorCode;
    }

    if (cursor.eid + 1 < leaf.getKeyCount())
    {
      cursor.eid += 1;
    }
    else
    {
      cursor.pid = leaf.getNextNodePtr();
      cursor.eid = 0;
      if (cursor.pid == -1)
      {
        return RC_END_OF_TREE;
      }
    }

    return 0;
}
